package com.ecloud.hbase.util;

import com.ecloud.common.ResultEnum;
import com.ecloud.common.exception.ServerException;
import com.ecloud.hbase.entities.HbaseCellData;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.util.Bytes;

import java.nio.charset.StandardCharsets;
import java.util.*;

import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * HbaseUtil
 *
 * @author Squbi
 * @date 2021/6/18 15:50
 */
public class HbaseUtil {
    private static final Logger logger = LoggerFactory.getLogger(HbaseUtil.class);

    private HbaseUtil() {
    }

    public static NavigableMap<byte[], byte[]> toNavigableMap(Map<String, String> map) {
        TreeMap<byte[], byte[]> treeMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
        for (Map.Entry<String, String> entry : map.entrySet()) {
            treeMap.put(Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue()));
        }
        return treeMap;
    }

    public static Map<String, String> toMap(NavigableMap<byte[], byte[]> map) {
        Map<String, String> targetMap = new HashMap<>(32);
        for (Map.Entry<byte[], byte[]> entry : map.entrySet()) {
            String sKey = new String(entry.getKey(), StandardCharsets.UTF_8);
            String sValue = new String(entry.getValue(), StandardCharsets.UTF_8);
            targetMap.put(sKey, sValue);
        }
        return targetMap;
    }

    public static Map<String, String> resultToMap(Result result, String familyName) {
        NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(familyName));
        if (familyMap == null || familyMap.isEmpty()) {
            logger.info("family is null : {}", familyName);
            throw new ServerException(ResultEnum.ARGS_ERR);
        }
        return toMap(familyMap);
    }

    public static List<HbaseCellData> resultToCellList(Result result) {
        Cell[] cells = result.rawCells();
        List<HbaseCellData> list = new ArrayList<>();
        for (Cell cell : cells) {
            HbaseCellData data = new HbaseCellData(
                    cell.getTimestamp(),
                    CellUtil.cloneFamily(cell),
                    CellUtil.cloneQualifier(cell),
                    CellUtil.cloneValue(cell)
            );
            list.add(data);
        }
        return list;
    }
}
